package com.qianfeng.reactor;

//
//                            _ooOoo_  
//                           o8888888o  
//                           88" . "88  
//                           (| -_- |)  
//                            O\ = /O  
//                        ____/`---'\____  
//                      .   ' \\| |// `.  
//                       / \\||| : |||// \  
//                     / _||||| -:- |||||- \  
//                       | | \\\ - /// | |  
//                     | \_| ''\---/'' | |  
//                      \ .-\__ `-` ___/-. /  
//                   ___`. .' /--.--\ `. . __  
//                ."" '< `.___\_<|>_/___.' >'"".  
//               | | : `- \`.;`\ _ /`;.`/ - ` : | |  
//                 \ \ `-. \_ __\ /__ _/ .-` / /  
//         ======`-.____`-.___\_____/___.-`____.-'======  
//                            `=---='  
//  
//         .............................................  
//                  佛祖镇楼                  BUG辟易  
//          佛曰:  
//                  写字楼里写字间，写字间里程序员；  
//                  程序人员写程序，又拿程序换酒钱。  
//                  酒醒只在网上坐，酒醉还来网下眠；  
//                  酒醉酒醒日复日，网上网下年复年。  
//                  但愿老死电脑间，不愿鞠躬老板前；  
//                  奔驰宝马贵者趣，公交自行程序员。  
//                  别人笑我忒疯癫，我笑自己命太贱；  
//  


import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.IOException;

/**
 * Created by Jackiechan on 2022/7/7 15:25
 *
 * @author Jackiechan
 * @version 1.0
 * @since 1.0
 */
public class TestFlux {

    public static void main(String[] args) throws IOException, InterruptedException {
        //创建发布者,就是提供数据的一方
        //a b c d就是生产出来的数据,就类似于我们从数据库中查询出来的数据,当然我们可以通过其他方式来获取这几个数据
//        Flux<String> flux = Flux.just("a", "b", "c", "d").doOnNext(s -> {
//            System.err.println("doOnext执行了");
//        }).doOnComplete(()->{
//            System.err.println("所有的数据都发送完成了");
//        });
//        //没有订阅就不会执行任何操作,直到我们订阅为止
//     //   flux.subscribe(System.err::println);
//        flux.subscribe(new Subscriber<String>() {
//
//            private Subscription subscription;
//            /**
//             * 当我们开始订阅的时候
//             * @param subscription
//             */
//            @Override
//            public void onSubscribe(Subscription subscription) {
//                this.subscription = subscription;
//                subscription.request(1);
//            }
//
//            /**
//             * 当我们拿到数据的时候
//             * @param s
//             */
//            @Override
//            public void onNext(String s) {
//                System.err.println("消费者收到了一个数据是:" + s);
//                //再给我来一个数据
//                subscription.request(1);
//            }
//
//            @Override
//            public void onError(Throwable throwable) {
//                System.err.println("消费者onError");
//            }
//
//            /**
//             * 当结束的时候
//             */
//            @Override
//            public void onComplete() {
//                System.err.println("消费者onComplete");
//            }
//        });

        //分割线===========
//        Flux.create(fluxSink -> {
//            //这一段代码就是我们要替换为业务的代码,就是数据得到
//            fluxSink.next("zhangsan").next("lisi").next("xujunfeng").next("nvpengyoujiehun").next("xinlangweizhi");
//        }).subscribe(System.out::println);
//
//        Flux.generate(synchronousSink -> {
//            ArrayList<Object> arrayList = new ArrayList<>();
//            for (int i = 0; i < 10; i++) {
//                arrayList.add("dasdas" + i);
//            }
//            //里面是一个集合也好,是一个对象也罢,甚至是一个字符串,对于它来说都是一个数据
//            synchronousSink.next(arrayList);
//          //  synchronousSink.next("xiaomei");
//            //通知结束,否则就是无限数据流,会一直给数据,并且是相同的数据
//            synchronousSink.complete();
//         }).subscribe(System.out::println);
//        Flux<String> just = Flux.just("shaokao", "xiaolongxia", "chaocai", "huamaoyiti", "paihuanggua");
//        Flux<String> from = Flux.from(just);
//        just.subscribe(System.out::println);
//        from.subscribe(System.err::println);
        //分割线 from array
//
//        Flux<String> flux = Flux.fromArray(new String[]{"rxjava", "thinking java", "efftive java", "java新特性", "JDK17", "Spring6"});
//        flux.subscribe(System.out::println);
//分割线,来自于可迭代的数据,
//        HashSet<String> hashSet = new HashSet<>();
//        for (int i = 0; i < 10; i++) {
//            hashSet.add(UUID.randomUUID().toString());
//        }
//        hashSet.stream().forEach(System.out::println);
//        //通过一个可迭代的数据来创建,比如set,list等都可以
//        Flux.fromIterable(hashSet).subscribe(System.err::println);

//        ArrayList<String> list = new ArrayList<>();
//        for (int i = 0; i < 20; i++) {
//            list.add("这是集合中的:" + i);
//        }
//        //将集合转成stream流
//        Stream<String> stream = list.stream().map(s -> s+"zhuanhuan");
//        //通过stream流来构建flux,这样集合的原始数据不变
//        Flux<String> flux = Flux.fromStream(stream);
//        //stream被关闭或者被使用或者flux重复subscribe都会报错
//        //stream.close();
//        //stream.forEach(System.err::println);
//       // stream.peek(System.err::println);
//      //  flux.subscribe(System.err::println);
//        flux.subscribe(System.err::println);

        //Flux.defer(()->Flux.just("123","abc","nihao","laowangdetou")).subscribe(System.err::println);

//        Flux.defer(()->Flux.create(fluxSink -> {
//            fluxSink.next("laowang").next("xiaomei").next("xiaohong").next("xiaotaohong");
//        }).doOnNext(o -> {
//            try {
//                Thread.sleep(new Random().nextInt(5000));
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e);
//            }
//        })).subscribe(System.err::println);
        //每个一秒生成一次数据,是一个数字,自增是数字
//        Flux.interval(Duration.of(1000, ChronoUnit.MILLIS)).subscribe(System.err::println);
//
//        try {
//            System.in.read();
//        } catch (IOException e) {
//            throw new RuntimeException(e);
//        }
        //返回错误的flux,让程序知道出现问题了
//    Flux.error(new RuntimeException("哈哈哈哈哈哈哈哈哈")).subscribe(System.err::println);
//        Flux.never().subscribe(System.out::println);
        //从start开始,连续取多少值,比如现在是从10开始,取一百个,最终的值是109
//        Flux.range(10, 100).subscribe(System.out::println);
        //concatWithValues是指的是在flux结束之后再拼接,必须是结束之后,所以我们需要通过complate来发出结束的信号,然后才会拼接数据
//        Flux.create(fluxSink -> fluxSink.next("laowang").next("detou").next("xiang").complete())
//                //在结束后拼接这些值再次通知
//                .concatWithValues("pi","qiu").subscribe(System.out::println);


//        Flux.concat(Flux.just("1", "2", "3"), Flux.just("a", "b", "c")).subscribe(System.out::println);
        //如果发生错误,会终止操作,因为完成或者错误都是终止符
//        Flux.concat(Flux.just(1, 2, 0)
//                        .doOnNext(integer -> System.out.println(100/integer))//doOnNext在执行next之前先执行的操作
//                , Flux.just(4, 5, 6)).subscribe(new Subscriber<Integer>() {
//            Subscription subscription;
//            @Override
//            public void onSubscribe(Subscription s) {
//                s.request(1);
//                subscription = s;
//            }
//
//            @Override
//            public void onNext(Integer integer) {
//                System.err.println(integer);
//                subscription.request(1);
//            }
//
//            @Override
//            public void onError(Throwable t) {
//                t.printStackTrace();
//            }
//
//            @Override
//            public void onComplete() {
//                System.err.println("TestMain中的方法onComplete 执行了");
//            }
//        });

        //如果发生问题,错误的数据会往后放,先处理下一个背压操作,最后再重新处理异常操作
//        Flux.concatDelayError(Flux.just(1, 2, 0)
//                        .doOnNext(integer -> System.out.println(100 / integer))//doOnNext在执行next之前先执行的操作
//                , Flux.just(4, 5, 6)).subscribe(new Subscriber<Integer>() {
//            Subscription subscription;
//
//            @Override
//            public void onSubscribe(Subscription s) {
//                s.request(1);
//                subscription = s;
//            }
//
//            @Override
//            public void onNext(Integer integer) {
//                System.err.println(integer);
//                subscription.request(1);
//            }
//
//            @Override
//            public void onError(Throwable t) {
//                t.printStackTrace();
//            }
//
//            @Override
//            public void onComplete() {
//                System.err.println("TestMain中的方法onComplete 执行了");
//            }
//        });

//        Flux.firstWithSignal(Flux.just(1, 2, 0) , Flux.just(4, 5, 6)).subscribe(new Subscriber<Integer>() {
//            Subscription subscription;
//            @Override
//            public void onSubscribe(Subscription s) {
//                s.request(1);
//                subscription = s;
//            }
//
//            @Override
//            public void onNext(Integer integer) {
//                System.err.println(integer);
//                subscription.request(1);
//            }
//
//            @Override
//            public void onError(Throwable t) {
//                t.printStackTrace();
//            }
//
//            @Override
//            public void onComplete() {
//                System.err.println("TestMain中的方法onComplete 执行了");
//            }
//        });

//        Flux<Long> flux = Flux.merge(Flux.interval(Duration.ofMillis(0), Duration.ofMillis(100)).take(5), Flux.interval(Duration.ofMillis(50), Duration.ofMillis(100)).take(5));
//     //   flux.toStream().forEach(System.out::println);
//        flux.subscribe(System.out::println);
//        System.in.read();
        //默认是按照数据的天然顺序比较的,如果是自定义的顺序,注意是flux之间的比较顺序
//        Flux.mergeComparing((s1,s2)->s2-s1,Flux.just(1,5,9),Flux.just(2,3,7),Flux.just(10))
//                .subscribe(System.out::println);
        //每个都是一个数字的时候就是刚好是我们想要的顺序
//        Flux.mergeComparing((s1, s2)->s2-s1,Flux.just(1),Flux.just(2),Flux.just(10),Flux.just(9))
//                .subscribe(System.out::println);
        //默认是从每个flux中拿到一个数据和另外flux中的数据放到一个数组中
//        Flux.zip(Flux.just(1,5,9,1),Flux.just(2,3,7,4)) .subscribe(System.out::println);
        //我们可以自定义一个拼接的方式,来决定返回什么数据
        //Flux.zip(Flux.just(1,5,9,1),Flux.just(2,3,7,4),(s1,s2)->s1+(s2+"")).subscribe(System.out::println);
        // 判断内部的数据是不是都符合断言的要求
        //  Flux.just(1, 2, 3).all(i -> i > 0).subscribe(System.out::println);
        //只要有一个数据符合条件就是true,除非全部都不符合
        //   Flux.just("1","2","3").any(s->!s.equals("")).subscribe(System.out::println);
        //将所有的数据挨个传递给我们,我们最终将数组进行响应的操作后返回一个结果,具体返回什么取决于我们自己的业务
//        String as = Flux.just(1, 2, 3, 4).as(new Function<Flux<Integer>, String>() {
//            @Override
//            public String apply(Flux<Integer> integerFlux) {
////                String[] s = {""};
////                integerFlux.subscribe(integer -> s[0] += integer);
////                return  s[0];
//                StringBuffer buffer = new StringBuffer("");
//                integerFlux.subscribe(integer -> buffer.append(integer));
//                buffer.append("结果");
//                return buffer.toString();
////                return "heheda";
//            }
//        });
//
//        System.err.println(as);
        //buffer一次缓冲多少数据,然后一次性作为一个数组返回,不足的数据将放到一个数组中
        //Flux.just(1,2,3,4,5).buffer(3).subscribe(System.out::println);
        //每次缓冲多少数据,作为一个数组的数据返回,每次会在原先的数据的技术上,跳过一个数据,再生成一个新的数组返回,类似于回声
        // 你还在吗? 还在吗? 在吗? 吗? ?
        // Flux.just(1,2,3,4,5).buffer(3,1).subscribe(System.out::println);
//        Flux<Integer> cache = Flux.just(1, 2, 3, 4, 5).cache(2);
//        cache.subscribe(System.out::println);
//        //第二次次订阅后只有缓存的最后的个数的数据
//        cache.subscribe(System.out::println);
        //将最后指定个数的数据缓存,保存一定时间,在这个时间内订阅的话得到的是缓存的数据,当缓存时间结束后,再次订阅得到的结果就是是原始数据
//        Flux<Integer> cache1 = Flux.just(1, 2, 3, 4, 5).cache(2, Duration.of(10, ChronoUnit.SECONDS));
//        cache1.subscribe(System.out::println);
//        System.out.println("-------");
//        cache1.subscribe(System.out::println);
//        System.out.println("-------");
//        cache1.subscribe(System.out::println);
//        System.out.println("-------");
//        cache1.subscribe(System.out::println);
//        System.out.println("-------");
//        Thread.sleep(11000);
//        //过期后再订阅就是原始数据
//        cache1.subscribe(System.out::println);

        //强制将数据转换成其他的类型,注意可能会出现类型转换异常
//        Flux<Integer> integerFlux = Flux.just(1, 2, 3, 4);
//        Flux<String> stringFlux = integerFlux.cast(String.class);
//        stringFlux.subscribe(System.out::println);
        //我们的flux的泛型和最终类型不一样,但是我们确定它就是最终类型的数据,这个时候我们进行cast是不会出现类型转换异常
//        Flux<Object> flux = Flux.just("14", "2", "3");
//        flux.cast(String.class).subscribe(System.out::println);
        //将数据保存到一个集合中,然后作为一个数据放到mono中
//        Flux.just(1, 2, 3, 4, 5, 6, 6).collect(Collectors.toSet()).subscribe(System.out::println);
//        Flux.just(1, 2, 3, 4, 5, 6, 6).collect(
//                ()->{
//                    //这个方法只会执行一次,而不是每次拿到一个数据的时候来执行一次
//                    if (new Random().nextBoolean()) {
//                        return "a";
//                    }
//                    return "b";
//                },//将这个返回的a拼接到我们的每个数据的前面
//                (t,u)->System.out.println(t+u)//t就是我们每次拼接到前面的那个a, u就是每次遍历的数据
//        ).subscribe(System.err::println);//订阅其实就是我们返回的那个a
        //将数据全部放到一个集合中之后作为一个返回值,放到mono中
       // Mono<List<Integer>> collectList = Flux.just(1, 2, 3, 4, 5, 6, 6).collectList();
       // collectList.subscribe(System.err::println);
        //Flux.empty().defaultIfEmpty("morenzhi").subscribe(System.err::println);
      //  Flux.just(1, 2, 3, 4, 5, 6, 6).defaultIfEmpty(100).subscribe(System.err::println);
        //去重
      //  Flux.just(1, 2, 3, 4, 5, 6, 6).distinct().subscribe(System.err::println);
        //连续的重复内容会去重,直到中间存在一个不一样的数据之后才会重新拿到数据
        //Flux.just(1, 2, 3, 4,4,4,4,4,4,4,4,4,4, 5,5,4,5,5,4).distinctUntilChanged().subscribe(System.out::println);
        // 参数1,用于判断我们的数据是不是已经存在了,如果有符合这个条件的数据就代表存在过
        //之前我们判断一个数据是不是存在是靠=判断,只有现在保存的数据中有=这个值才认为存在,现在我们自己改了一个判断方式
        //比如当前的例子,1不大于6,保存起来, 2不大于6,但是因为已经有一个符合不大于6的数据在里面, 所以2就不会保存.依次类推,直到8大于6,保存起来,依次类推,到10大于6但是已经有8了,所以不保存,以此类推
       // Flux.just(1, 2, 3, 4, 8,5, 6, 6,10,100).distinct(integer -> integer>6,()->new HashSet<>()).subscribe(System.err::println);

       // Flux.just(1, 2, 3, 4, 8,5, 6, 6,10,100).filter(integer -> integer>5).subscribe(System.err::println);
        //和java中的stream的map是一样的,并不是数据本身发生了改变,而是通过一些方法对数据进行操作,返回了一个新的类型的数据
        //Flux.just(1, 2, 3, 4, 8, 5, 6, 6, 10, 100).map(integer -> integer + "转换").subscribe(System.err::println);
        // 让前面的flux结束,然后执行后面提供的mono中的操作,flux没有人订阅获取数据,但是实际上可以认为是被订阅了,只是没有具体处理措施,doOnNext是执行的
        Flux.just(1, 2, 3, 4, 8, 5, 6, 6, 10, 100).doOnNext(integer -> {
            System.err.println("当前拿到了" + integer);
        }).doOnComplete(()->{
            System.err.println("结束了");
        }).then(Mono.just("abc")).subscribe(System.err::println);
    }
}
